package com.zk.iotgateway.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 *
 * 消费者工具类
 * @author:mn.m
 * @create 2019-04-29 上午 10:56
/* **/
public class ConsumerCreator implements Runnable{

    private IotRabbitCreator iotRabbitCreator;
    private String queueName;
    public ConsumerCreator(IotRabbitCreator iotRabbitCreator,String queueName){
        this.iotRabbitCreator=iotRabbitCreator;
        this.queueName=queueName;
    }

    @Override
    public  void run() {

            try {
                Channel channel =iotRabbitCreator.creatRabbitChannel(null,8092,null,null,null);
                //这里重写了DefaultConsumer的handleDelivery方法，因为发送的时候对消息进行了getByte()，在这里要重新组装成String
                Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        super.handleDelivery(consumerTag, envelope, properties, body);
                        String message = new String(body, "UTF-8");
                        try {
                            /*对应业务类处理*/

                            //消息确认
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }catch (Exception e){
                            //出现异常，置为true，重新入队。
                             channel.basicAck(envelope.getDeliveryTag(), true);
                            /*出现异常，不重新入队，而是reject入死信队列。
                             channel.basicReject(envelope.getDeliveryTag(), false);*/
                        }
                    }
                };
                //声明队列中被消费掉的消息（参数为：队列名称；消息是否自动确认;consumer主体）
                channel.basicConsume(queueName, false, consumer);
                //这里不能关闭连接，调用了消费方法后，消费者会一直连接着rabbitMQ等待消费
            } catch (IOException e) {
                e.printStackTrace();
            }

    }
}
